package com.phoenix.rabbitmq.service.consumer;


import com.phoenix.rabbitmq.config.RabbitMQConfig;
import com.phoenix.rabbitmq.pojo.OrderMessage;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;


@Component
public class OrderConsumer {
    private static final Logger logger = LoggerFactory.getLogger(OrderConsumer.class);

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    /**
     * 订单消息消费者 - 手动ack实现
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderMessage(OrderMessage orderMessage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        String redisKey =  "order:msg:" + orderMessage.getOrderId();
        try{
            //检查消息是否已处理(幂等性控制)
            if(redisTemplate.hasKey(redisKey)){
                logger.warn("⚠️ 消息已处理，直接ACK，订单ID：{}", orderMessage.getOrderId());
                channel.basicAck(deliveryTag,false);
                return;
            }
            //处理业务逻辑
            logger.info("✅ 开始处理订单消息：{}", orderMessage);
            processOrder(orderMessage);

            //将处理成功的消息id存入redis，设置过期时间防止内存泄漏
            redisTemplate.opsForValue().set(redisKey,"processed",30,TimeUnit.MINUTES);

            //手动确认消息ack
            channel.basicAck(deliveryTag,false);
            logger.info("🎉 订单处理完成并ACK确认，订单ID：{}", orderMessage.getOrderId());

        } catch (Exception e) {
            logger.error("❌ 订单处理异常，订单ID：{}，异常信息：", orderMessage.getOrderId(), e);
            try{
                //处理失败，拒绝消息并重新入队(根据业务需求调整)
                if(getRetryCount(redisKey) < 3){
                    //重试次数未超限，重新入队
                    channel.basicNack(deliveryTag,false,true);
                    incrementRetryCount(redisKey);
                    logger.info("🔄 消息重新入队，订单ID：{}，重试次数：{}",
                            orderMessage.getOrderId(), getRetryCount(redisKey));
                }else{
                    // 重试次数超限，拒绝并不再重新入队
                    channel.basicNack(deliveryTag, false, false);
                    logger.error("🚫 消息重试超限，转入死信队列，订单ID：{}", orderMessage.getOrderId());
                }
            } catch (IOException ex) {
                logger.error("❌ ACK操作异常：", ex);
            }
        }

    }

    /**
     * 模拟订单处理业务逻辑
     */
    private void processOrder(OrderMessage orderMessage) throws InterruptedException {
        //模拟业务处理
        Thread.sleep(1000);
        // 模拟可能的业务异常
        if ("error".equals(orderMessage.getProductName())) {
            throw new RuntimeException("模拟业务处理异常");
        }

        // 实际业务处理逻辑...
        logger.info("处理订单成功：订单ID[{}]，金额[{}]",
                orderMessage.getOrderId(), orderMessage.getAmount());
    }

    /**
     * 获取消息重试次数
     */
    private int getRetryCount(String redisKey) {
        String count = redisTemplate.opsForValue().get(redisKey + ":retry_count");
        return count == null ? 0 : Integer.parseInt(count);
    }

    /**
     * 增加重试次数
     */
    private void incrementRetryCount(String redisKey) {
        int currentCount = getRetryCount(redisKey);
        redisTemplate.opsForValue().set(redisKey + ":retry_count",
                String.valueOf(currentCount + 1), 30, TimeUnit.MINUTES);
    }

}
